package com.lzwk.app.ods.mdb;

import com.alibaba.fastjson.JSON;
import com.lzwk.beans.ocpx.AdbClassStudent;
import com.lzwk.beans.ocpx.AdbStudentInfo;
import com.lzwk.beans.ocpx.DimDivosion;
import com.lzwk.beans.ocpx.WkFormAccountInfo;
import com.lzwk.common.ocpx.OCPXConstant;
import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: MdbJdbcTables
 * @Author Yeaser
 * @Date: 2022/2/18 19:00
 * @Version 1.0
 * Description: TODO
 */
public class MdbLoookupTables {


    public static void ocpxDimDivision(StreamTableEnvironment tenv, SingleOutputStreamOperator<AdbClassStudent> sourceClassStudent, SingleOutputStreamOperator<WkFormAccountInfo> sourceWkAccountInfo) {


        KeyedStream<AdbClassStudent, Integer> adbClassStudentIntegerKeyedStream = sourceClassStudent
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<AdbClassStudent>forBoundedOutOfOrderness(Duration.ofDays(1))
                .withTimestampAssigner((data,ts) -> data.getCreate_time().getTime()))
                .keyBy(AdbClassStudent::getAccount_id);




        KeyedStream<WkFormAccountInfo, Integer> wkFormAccountInfoIntegerKeyedStream = sourceWkAccountInfo
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WkFormAccountInfo>forBoundedOutOfOrderness(Duration.ofDays(1))
                .withTimestampAssigner((data,ts) -> data.getCreate_time().getTime()))
                .keyBy(WkFormAccountInfo::getId);


        SingleOutputStreamOperator<AdbStudentInfo> adbStudentInfoDS = adbClassStudentIntegerKeyedStream
                .intervalJoin(wkFormAccountInfoIntegerKeyedStream)
                .between(Time.days(-180), Time.days(180))
                .process(new ProcessJoinFunction<AdbClassStudent, WkFormAccountInfo, AdbStudentInfo>() {
                    @Override
                    public void processElement(AdbClassStudent left, WkFormAccountInfo right, Context ctx, Collector<AdbStudentInfo> out) throws Exception {
                        out.collect(new AdbStudentInfo(left, right));
                    }
                });




        Table adbStudentInfoTable = tenv.fromDataStream(adbStudentInfoDS,
                $("account_id"), $("term_class_id"), $("division_activity_id"), $("activity_term_id"), $("state"), $("refund_status"),
                $("remark"), $("create_time").rowtime(), $("update_time"), $("op"),  $("id"), $("gender"),
                $("telephone"), $("career"), $("age"), $("wechat_nickname"), $("pt").proctime());

        tenv.createTemporaryView("adb_division_class_student",adbStudentInfoTable);

        /*adbStudentInfoDS.print();*/

/*        Table table1 = tenv.fromDataStream(sourceClassStudent,
                $("account_id"), $("term_class_id"), $("division_activity_id"), $("activity_term_id"), $("state"), $("refund_status"),
                $("remark"), $("create_time").rowtime(), $("update_time"), $("id"), $("op"), $("pt").proctime());

        Table table2 = tenv.fromDataStream(sourceWkAccountInfo,
                $("id").as("wk_account_id"), $("full_name"), $("gender"), $("age"), $("id_card"), $("telephone"), $("email"), $("phone"),
                $("career"), $("address"), $("wechat_account"), $("wechat_nickname"), $("main_address"), $("backup_1"), $("backup_2"),
                $("backup_3"), $("state").as("account_state"), $("create_time").rowtime().as("account_time"), $("update_time").as("account_update_time"),
                $("remark").as("account_remark"), $("pt").proctime().as("account_pt"));*/



        tenv.executeSql("create table dim_division_tmp(" +
                "   `account_id` INT, " +
                "   `class_id` INT, " +
                "   `parent_activity_id` INT, " +
                "   `activity_id` INT, " +
                "   `parent_term_id` INT, " +
                "   `term_id` INT, " +
                "   `student_state` STRING, " +
                "   `refund_status` INT, " +
                "   `student_create_time` STRING, " +
                "   `liveroom_id` INT, " +
                "   `target_roi` double, " +
                "   `activity_name` STRING, " +
                "   `activity_state` STRING, " +
                "   `term_rank` INT," +
                "   `term_start_time` STRING," +
                "   `term_end_time` STRING," +
                "   `term_student_cnt` INT," +
                "   `term_state` STRING, " +
                "   `term_promote_start_time` STRING, " +
                "   `class_channel_category` STRING, " +
                "   `class_name` STRING, " +
                "   `class_rank` INT, " +
                "   `class_object_id` INT, " +
                "   `class_object_type` STRING, " +
                "   `class_student_number` INT, " +
                "   `class_can_redirect` INT, " +
                "   `class_state` STRING, " +
                "   `class_monitor_id` INT, " +
                "   `class_monitor_name` STRING, " +
                "   `age` STRING, " +
                "   `career` STRING, " +
                "   `city` STRING, " +
                "   `headimgurl` STRING, " +
                "   `city_level` STRING, " +
                "   `sex` STRING, " +
                "   `province` STRING, " +
                "   `create_time` STRING, " +
                "   `source_type` STRING, " +
                "   `nickname` STRING, " +
                "   `input_cost` double, " +
                "   `phone_num` STRING, " +
                "   `form_acc_sex` INT, " +
                "   `op` INT, " +
                "   PRIMARY KEY (`account_id`, `class_id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "." + OCPXConstant.OCPXDIMDIVISION +"', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")");



        tenv.executeSql(" insert into dim_division_tmp " +
                " select `account_id`, " +
                "        `class_id`, " +
                "        `parent_activity_id`, " +
                "        `activity_id`, " +
                "        `parent_term_id`, " +
                "        `term_id`, " +
                "        `student_state`, " +
                "        `refund_status`, " +
                "        student_create_time, " +
                "        `liveroom_id`, " +
                "        `target_roi`, " +
                "        `activity_name`, " +
                "        `activity_state`, " +
                "        `term_rank`, " +
                "        `term_start_time`, " +
                "        `term_end_time`, " +
                "        `term_student_cnt`, " +
                "        `term_state`, " +
                "        `term_promote_start_time`, " +
                "        `class_channel_category`, " +
                "        `class_name`, " +
                "        `class_rank`, " +
                "        `class_object_id`, " +
                "        `class_object_type`, " +
                "        `class_student_number`, " +
                "        `class_can_redirect`, " +
                "        `class_state`, " +
                "        `class_monitor_id`, " +
                "        `class_monitor_name`, " +
                "        `age`, " +
                "        `career`, " +
                "        `city`, " +
                "        `headimgurl`, " +
                "        `city_level`, " +
                "        `sex`, " +
                "        `province`, " +
                "        `create_time`, " +
                "        `source_type`, " +
                "        `nickname`, " +
                "        `input_cost`, " +
                "        `phone_num`, " +
                "        `form_acc_sex`, " +
                "        `op` " +
                "        from( " +
                " SELECT ifnull(student.`account_id`, -1) AS `account_id`, " +
                "       ifnull(student.`term_class_id`,-1)                                                    AS class_id, " +
                "       ifnull(activity.`parent_id`,-1)                                                       AS `parent_activity_id`, " +
                "       ifnull(student.`division_activity_id`,-1)                                             AS activity_id, " +
                "       ifnull(term.`parent_id`,-1)                                                           AS `parent_term_id`, " +
                "       ifnull(student.`activity_term_id`,-1)                                                 AS term_id, " +
                "       ifnull(regexp_replace(student.`state`,'\\n|\\t',''),'')                                AS student_state, " +
                "       ifnull(student.`refund_status`,-1)                                                    AS refund_status, " +
                "       DATE_FORMAT(ifnull(student.`create_time`,TO_TIMESTAMP('1970-01-01 08:00:01')),'yyyy-MM-dd HH:mm:ss')                     AS student_create_time, " +
                "       ifnull(activity.`liveroom_id`, -1)                                                    AS `liveroom_id`, " +
                "       ifnull(regexp_replace(activity.`activity_name`,'\\n|\\t',''),'')                      AS `activity_name`, " +
                "       ifnull(regexp_replace(activity.`state`,'\\n|\\t',''),'')                              AS activity_state, " +
                "       ifnull(term.`rank`,-1)                                                                AS term_rank, " +
                "       DATE_FORMAT(ifnull(term.`start_time`,TO_TIMESTAMP('1970-01-01 08:00:01')),'yyyy-MM-dd HH:mm:ss')                         AS term_start_time, " +
                "       DATE_FORMAT(ifnull(term.`end_time`,TO_TIMESTAMP('1970-01-01 08:00:01')),'yyyy-MM-dd HH:mm:ss')                           AS `term_end_time`, " +
                "       ifnull(term.`student_cnt`,-1)                                                         AS term_student_cnt, " +
                "       ifnull(term.`state`,'')                                                               AS term_state, " +
                "       DATE_FORMAT(ifnull(term.`promote_start_time`,TO_TIMESTAMP('1970-01-01 08:00:01')),'yyyy-MM-dd HH:mm:ss')                 AS `term_promote_start_time`, " +
                "       ifnull(regexp_replace(class.`channel_category`,'\\n|\\t',''),'')                      AS class_channel_category, " +
                "       ifnull(regexp_replace(class.`class_name`,'\\n|\\t',''),'')                            AS class_name, " +
                "       ifnull(class.`rank`,-1)                                                               AS class_rank, " +
                "       ifnull(class.`object_id`,-1)                                                          AS class_object_id, " +
                "       ifnull(regexp_replace(class.`object_type`,'\\n|\\t',''),'')                           AS class_object_type, " +
                "       ifnull(class.`student_number`,-1)                                                     AS class_student_number, " +
                "       ifnull(class.`can_redirect`,0)                                                        AS class_can_redirect, " +
                "       ifnull(regexp_replace(class.`state`,'\\n|\\t',''),'')                                 AS class_state, " +
                "       ifnull(class.`monitor_id`,-1)                                                         AS `class_monitor_id`, " +
                "       ifnull(regexp_replace(class.`monitor_name`,'\\n|\\t',''),'')                          AS `class_monitor_name`, " +
                "       ifnull(student.age,'')                                                                AS age, " +
                "       ifnull(regexp_replace(student.career,'\\n|\\t',''),'')                                AS career, " +
                "       ifnull(regexp_replace(acc.city,'\\n|\\t',''),'')                                      AS city, " +
                "       ifnull(regexp_replace(acc.headimgurl,'\\n|\\t',''),'')                                AS headimgurl, " +
                "       case " +
                "           when acc.city in ('北京', '上海', '广州', '深圳') then '一线城市' " +
                "           when acc.city in ('成都', '杭州', '重庆', '武汉', '西安', '苏州', '天津', '南京', '长沙', '郑州', '东莞', '青岛', '沈阳', '宁波', '昆明') " +
                "               then '新一线城市' " +
                "           when acc.city in " +
                "                ('无锡', '佛山', '合肥', '大连', '福州', '厦门', '哈尔滨', '济南', '温州', '南宁', '长春', '泉州', '石家庄', '贵阳', '南昌', '金华', '常州', " +
                "                 '南通', '嘉兴', '太原', '徐州', '惠州', '珠海', '中山', '台州', '烟台', '兰州', '绍兴', '海口', '扬州') then '二线城市' " +
                "           else '其他城市' " +
                "           end                                                                               AS city_level, " +
                "       ifnull(cast(student.gender AS STRING), acc.sex)                                      AS sex,  " + //"性别:未知0 男1 女2"
                "       ifnull(regexp_replace(acc.province,'\\n|\\t',''),'')                                  AS province, " +
                "       DATE_FORMAT(ifnull(acc.create_time,TO_TIMESTAMP('1970-01-01 08:00:01')),'yyyy-MM-dd HH:mm:ss')                           AS create_time, " +
                "       GetJsonString(ifnull(regexp_replace(student.remark,'\\n|\\t',''),''), 'source_type')    AS source_type, " +
                "       if(regexp_replace(student.wechat_nickname,'\\n|\\t','') <> '', regexp_replace(student.wechat_nickname,'\\n|\\t',''), regexp_replace(acc.nickname,'\\n|\\t',''))            AS nickname, " +
                "       if(student.telephone <> '', ifnull(regexp_replace(student.telephone,'\\n|\\t',''),''), ifnull(regexp_replace(acc.telephone,'\\n|\\t',''),''))                       AS phone_num, " +
                "       ifnull(cast(student.gender as INT),0)                                                AS form_acc_sex, " +
                "       if(ifnull(student.`op`,true)=true,1,0)                                                AS op, " +
                "       ifnull(GetJsonDouble(ifnull(term.remark,''), 'input_cost'), 0.00)                                  AS input_cost, " +
                "       ifnull(GetJsonDouble(ifnull(term.remark,''), 'target_roi'), 0.00)                                  AS target_roi " +
                "       FROM adb_division_class_student student " +
                "       JOIN mdb_account_account  FOR SYSTEM_TIME AS OF `student`.`pt` AS acc " +
                "              ON acc.id = student.account_id " +
                "       JOIN adb_division_term_class FOR SYSTEM_TIME AS OF `student`.`pt` AS class " +
                "              ON student.term_class_id = class.id " +
                "              AND class.object_type = 'channel' " +
                "       JOIN adb_division_activity_term FOR SYSTEM_TIME AS OF `student`.`pt` AS term " +
                "              ON student.activity_term_id = term.id " +
                "       JOIN adb_division_activity FOR SYSTEM_TIME AS OF `student`.`pt` AS activity " +
                "              ON student.division_activity_id = activity.id " +  //用户 取消授权/重新授权场景,归档表和学员表会有相同学员记录,需要根据归档时间>授权时间记录来判断是否被删除
                ")");


/*        tenv
                .toRetractStream(table, DimDivosion.class)
                .filter(data -> data.f0)
                .map(data -> JSON.toJSONString(data.f1) )
                .addSink(MyKafkaUtil.getKafkaProducer(OCPXConstant.OCPXTESTTOPIC2,OCPXConstant.TESTKAFKABROKERS));*/






    }
}
